package ct.consumer.coprocessor;

import ct.common.bean.BaseDao;
import ct.common.constant.Names;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @Author Jeremy Zheng
 * @Date 2021/2/25 12:04
 * @Version 1.0
 * 使用协处理器保存被叫用户的数据
 *
 * 协处理器的使用
 * 1.创建类
 * 2.让表知道协处理类
 * 3.将协处理项目打成jar包发布到hbase中（关联的jar包也要发布），并且需要分发到集群的每一个节点
 */
public class InsertCalleeCoprocessor extends BaseRegionObserver {

    //主叫用户数据保存完后触发操作：hbase自动保存被叫用户数据
    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {

        //获取表
        Table table = e.getEnvironment().getTable(TableName.valueOf(Names.TABLE.getValue()));

        //主叫用户数据的rowkey
        String rowkey = Bytes.toString(put.getRow());
        String[] values = rowkey.split("_");

        CoprocessorDao dao = new CoprocessorDao();
        String call2 = values[1];
        String call1 = values[3];
        String callTime = values[2];
        String duration = values[4];
        String flg = values[5];

        //只有主叫用户数据被保存才触发被叫用户数据的保存，
        //被叫用户数据的保存不允许再触发又一次的被叫用户数据的保存（不然会死循环）
        if ("1".equals(flg)) {
            String calleeRowkey = dao.getRegionNum(call2, callTime)
                    + "_" + call2 + "_" + callTime + "_" + call1 + "_" + duration + "_0";
            byte[] calleeFamily = Bytes.toBytes(Names.CF_CALLEE.getValue());
            Put calleePut = new Put(Bytes.toBytes(calleeRowkey));

            calleePut.addColumn(calleeFamily, Bytes.toBytes("call1"), Bytes.toBytes(call2));
            calleePut.addColumn(calleeFamily, Bytes.toBytes("call2"), Bytes.toBytes(call1));
            calleePut.addColumn(calleeFamily, Bytes.toBytes("callTime"), Bytes.toBytes(callTime));
            calleePut.addColumn(calleeFamily, Bytes.toBytes("duration"), Bytes.toBytes(duration));
            calleePut.addColumn(calleeFamily, Bytes.toBytes("flg"), Bytes.toBytes("0"));
            table.put(calleePut);

            table.close();
        }
    }

    private class CoprocessorDao extends BaseDao {
        public int getRegionNum(String tel, String time) {
            return genRegionNum(tel, time);
        }
    }

}
